package histogram;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import global.*;

public class Histogram extends BaseDriver {

	public Histogram(String[] args) {
		super(args, HistConsts.jobName);
		
	}

	public static void main(String[] args) {
		new Histogram(args);
	}

	@Override
	protected ArrayList<ArgumentParam> fillParamsList() {
		ArrayList<ArgumentParam> res = new ArrayList<ArgumentParam>(4);
		
		ArgumentParam arg;
		
		arg = new ArgumentParam(GlobalConsts.paramNameInputPath, GlobalConsts.defaultRelationsPath, ArgumentType.INPUT_FOLDER, false);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameOutputPath, GlobalConsts.defaultOutputPath, ArgumentType.OUTPUT_FOLDER, false);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameBinSizeR0, GlobalConsts.defaultBinSizeR0 + "", ArgumentType.GENERAL_INT, true);
		res.add(arg);
		
		arg = new ArgumentParam(GlobalConsts.paramNameBinSizeR1, GlobalConsts.defaultBinSizeR1 + "", ArgumentType.GENERAL_INT, true);
		res.add(arg);
		
		return res;
	}

	@Override
	protected JobConf createJobConf() {
		JobConf conf = new JobConf(Histogram.class);
		conf.setOutputKeyClass(MyTripleKey.class);
		conf.setOutputValueClass(IntWritable.class);
		
		conf.setMapperClass(HistogramMapper.class);
		conf.setCombinerClass(HistogramReducer.class);
		conf.setReducerClass(HistogramReducer.class);
		conf.setPartitionerClass(HistogramPartitioner.class);
		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		conf.setNumReduceTasks(2);
		
		return conf;
	}
	
	@Override
	protected void doPreProcess() {
		System.out.println("Deleting histograms folder...");
		FileSystem hdfs = null;
		try {
			hdfs =  FileSystem.get(getJobConf());
			try {
				Path path = new Path(GlobalConsts.defaultHistogramPath);
				hdfs.delete(path, true);
			}
			finally {
				hdfs.close();
			}
		}
		catch (Exception ex) {
			System.out.println("Warning: Could not delete histograms folder!");
			System.out.println(ex.getMessage());
		}
	}
	
	@Override
	protected void doAfterProcess() {
		System.out.println("Moving results to histograms folder...");
		FileSystem hdfs = null;
		try {
			hdfs =  FileSystem.get(getJobConf());
			try {
				Path pathFrom, pathTo;
				pathTo = new Path(GlobalConsts.defaultHistogramPath);
				hdfs.mkdirs(pathTo);
				boolean isFirstFileH0 = true;
				
				pathFrom = new Path(getOutputFolder() + "/" + GlobalConsts.getOutputFileName(0));
				if (!hdfs.exists(pathFrom)) {
					System.out.println("Error: " + pathFrom.toString() + " file not found!");
					System.exit(1);
				}
				BufferedReader inStream = new BufferedReader(new InputStreamReader(hdfs.open(pathFrom)));
				try {
					char[] c = new char[1];
					inStream.read(c);
					isFirstFileH0 = c[0] == GlobalConsts.dataSetId0 + '0';
				}
				finally {
					inStream.close();
				}
				
				if (isFirstFileH0) {
					pathTo = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram0);
				}
				else {
					pathTo = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram1);
				}
				
				hdfs.rename(pathFrom, pathTo);
				pathFrom = new Path(getOutputFolder() + "/" + GlobalConsts.getOutputFileName(1));
				if (!hdfs.exists(pathFrom)) {
					System.out.println("Error: " + pathFrom.toString() + " file not found!");
					System.exit(1);
				}
				if (isFirstFileH0) {
					pathTo = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram1);
				}
				else {
					pathTo = new Path(GlobalConsts.defaultHistogramPath + "/" + GlobalConsts.fileNameHistogram0);
				}
				
				hdfs.rename(pathFrom, pathTo);
			}
			finally {
				hdfs.close();
			}
		}
		catch (Exception ex) {
			System.out.println("Warning: Could not move results to histograms folder!");
			System.out.println(ex.getMessage());
		}
	}

	@Override
	protected void printExtraUsage() {
		System.out.println("Note: <" + GlobalConsts.paramNameInputPath + "> value should contain both " +
				GlobalConsts.fileNameRelation0 + " and " +
				GlobalConsts.fileNameRelation1);
	}
}
